package com.jiayuan.cn.energy.test;

//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.shaded.netty4.io.netty.handler.codec.mqtt.MqttProperties;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//import org.eclipse.paho.client.mqttv3.MqttTopic;

//import com.github.davidfantasy.flink.connector.mqtt.MqttMessage;
//import com.github.davidfantasy.flink.connector.mqtt.MqttProperties;
//import com.github.davidfantasy.flink.connector.mqtt.MqttTopic;
//import com.github.davidfantasy.flink.connector.mqtt.source.MqttTableSourceFactory;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;

public class FlinkMqttStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create a DataStream
        DataStream<Row> dataStream = env.fromElements(
                Row.of("Alice", 12),
                Row.of("Bob", 10),
                Row.of("Alice", 100));

        // interpret the insert-only DataStream as a Table
        Table inputTable = tableEnv.fromDataStream(dataStream).as("name", "score");

        // register the Table object as a view and query it
        // the query contains an aggregation that produces updates
        tableEnv.createTemporaryView("InputTable", inputTable);
        Table resultTable = tableEnv.sqlQuery(
                "SELECT name, SUM(score) FROM InputTable GROUP BY name");
        // interpret the updating Table as a changelog DataStream
        DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);
        // add a printing sink and execute in DataStream API
        resultStream.print();
        env.execute();
    }

    public static void main0(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        MqttProperties mqttProp = new MqttProperties();
//        mqttProp.setHost("192.168.110.88");
//        mqttProp.setPort(1883);
//        mqttProp.setUsername("s3");
//        mqttProp.setPassword("111");
//        List<MqttTopic> topics = new ArrayList<>();
//        topics.add(new MqttTopic("/v1/report/+/model", 0));
//        var source = env.fromSource(new MqttTableSourceFactory(mqttProp, topics), WatermarkStrategy.noWatermarks(), "Mqtt Source");
//        source.map(v -> {
//            var msg = (MqttMessage) v;
//            return msg.getTopic() + ":" + new String(msg.getPayload());
//        }).print();
//        env.execute("MQTT Source Test");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 定义输出数据源
        String createSinkTableProduct = """
                CREATE TABLE test2 (
                        str STRING
                    ) WITH (
                        'connector'  = 'jdbc',
                        'url'        = 'jdbc:mysql://192.168.110.88:3306/helm?useUnicode=true&characterEncoding=utf-8&useSSL=false',
                        'table-name' = 'test2',
                        'driver'     = 'com.mysql.cj.jdbc.Driver',
                        'username'   = 'helm',
                        'password'   = 'db-999'
                    )
                """;
        tableEnv.executeSql(createSinkTableProduct);

        // 定义输入数据源
        String createSourceTableDdl = """
                CREATE TABLE mqttTest (
                    id INTEGER,
                     equipment_code  STRING
                ) WITH (
                    'connector' = 'mqtt',
                    'server' = '192.168.110.88',
                    'username' = 's3',
                    'password' = '111',
                    'port' = '1883',
                    'topic' = '/v1/report/+/model'
                )
                """;
        tableEnv.executeSql(createSourceTableDdl);

        String query = "INSERT INTO test2 SELECT equipment_code as str  FROM mqttTest ";
        tableEnv.executeSql(query);
    }
}